//! # 流处理器单元测试
//! 
//! 测试各种流处理器的功能和性能

use std::time::Duration;
use std::sync::Arc;
use tokio::time::sleep;
use serde_json::json;

use rustcloud_stream::prelude::*;
use rustcloud_bus::memory::InMemoryMessageBus;
use rustcloud_bus::message::MessageBuilder;

#[tokio::test]
async fn test_map_processor() {
    let processor = MapProcessor::with_mapper(|mut msg: Message| {
        let mut payload = msg.payload().clone();
        payload["processed"] = json!(true);
        payload["timestamp"] = json!(chrono::Utc::now().timestamp());
        msg.set_payload(payload);
        msg
    });

    let input = MessageBuilder::new()
        .with_topic("test.input")
        .with_payload(json!({"value": 42, "name": "test"}))
        .build();

    let result = processor.process(input).await.unwrap();
    
    assert_eq!(result.payload()["value"], json!(42));
    assert_eq!(result.payload()["name"], json!("test"));
    assert_eq!(result.payload()["processed"], json!(true));
    assert!(result.payload()["timestamp"].is_number());
    assert_eq!(processor.name(), "map");
}

#[tokio::test]
async fn test_filter_processor() {
    let processor = FilterProcessor::with_predicate(|msg: &Message| {
        msg.payload()["active"].as_bool().unwrap_or(false)
    });

    // 测试通过过滤器的消息
    let active_msg = MessageBuilder::new()
        .with_topic("test")
        .with_payload(json!({"active": true, "id": 1}))
        .build();

    let result = processor.process(active_msg).await.unwrap();
    assert!(result.is_some());
    assert_eq!(result.unwrap().payload()["id"], json!(1));

    // 测试被过滤掉的消息
    let inactive_msg = MessageBuilder::new()
        .with_topic("test")
        .with_payload(json!({"active": false, "id": 2}))
        .build();

    let result = processor.process(inactive_msg).await.unwrap();
    assert!(result.is_none());

    // 测试没有 active 字段的消息（默认为 false）
    let no_active_msg = MessageBuilder::new()
        .with_topic("test")
        .with_payload(json!({"id": 3}))
        .build();

    let result = processor.process(no_active_msg).await.unwrap();
    assert!(result.is_none());
}

#[tokio::test]
async fn test_aggregate_processor() {
    let processor = AggregateProcessor::new(
        "count_by_user",
        |msg: &Message| -> StreamResult<String> {
            Ok(msg.payload()["user_id"].as_str().unwrap_or("unknown").to_string())
        },
        |msg: &Message| -> StreamResult<i64> {
            Ok(msg.payload()["amount"].as_i64().unwrap_or(0))
        },
        |acc: &i64, value: &i64| acc + value,
        0i64,
    );

    // 第一个用户的消息
    let msg1 = MessageBuilder::new()
        .with_topic("transactions")
        .with_payload(json!({"user_id": "user1", "amount": 100}))
        .build();

    let result1 = processor.process(msg1).await.unwrap();
    assert_eq!(*result1.get("user1").unwrap(), 100);

    // 同一用户的另一个消息
    let msg2 = MessageBuilder::new()
        .with_topic("transactions")
        .with_payload(json!({"user_id": "user1", "amount": 50}))
        .build();

    let result2 = processor.process(msg2).await.unwrap();
    assert_eq!(*result2.get("user1").unwrap(), 150);

    // 不同用户的消息
    let msg3 = MessageBuilder::new()
        .with_topic("transactions")
        .with_payload(json!({"user_id": "user2", "amount": 75}))
        .build();

    let result3 = processor.process(msg3).await.unwrap();
    assert_eq!(*result3.get("user1").unwrap(), 150);
    assert_eq!(*result3.get("user2").unwrap(), 75);
}

#[tokio::test]
async fn test_batch_processor() {
    let processor = BatchProcessor::new("test_batch", 3, Duration::from_millis(100));

    // 发送前两个消息，不应该触发批处理
    let msg1 = MessageBuilder::new()
        .with_topic("test")
        .with_payload(json!({"id": 1}))
        .build();

    let result1 = processor.process(msg1).await.unwrap();
    assert!(result1.is_none());

    let msg2 = MessageBuilder::new()
        .with_topic("test")
        .with_payload(json!({"id": 2}))
        .build();

    let result2 = processor.process(msg2).await.unwrap();
    assert!(result2.is_none());

    // 第三个消息应该触发批处理
    let msg3 = MessageBuilder::new()
        .with_topic("test")
        .with_payload(json!({"id": 3}))
        .build();

    let result3 = processor.process(msg3).await.unwrap();
    assert!(result3.is_some());
    
    let batch = result3.unwrap();
    assert_eq!(batch.len(), 3);
    assert_eq!(batch[0].payload()["id"], json!(1));
    assert_eq!(batch[1].payload()["id"], json!(2));
    assert_eq!(batch[2].payload()["id"], json!(3));
}

#[tokio::test]
async fn test_route_processor() {
    let processor = RouteProcessor::new(
        "topic_router",
        |msg: &Message| {
            let msg_type = msg.payload()["type"].as_str().unwrap_or("default");
            format!("output.{}", msg_type)
        },
    );

    let msg = MessageBuilder::new()
        .with_topic("input")
        .with_payload(json!({"type": "error", "message": "Something went wrong"}))
        .build();

    let result = processor.process(msg.clone()).await.unwrap();
    assert_eq!(result.route, "output.error");
    assert_eq!(result.message.payload()["message"], json!("Something went wrong"));
}

#[tokio::test]
async fn test_transform_processor() {
    let processor = TransformProcessor::new(
        "user_transformer",
        |msg: Message| -> StreamResult<serde_json::Value> {
            let payload = msg.payload();
            let transformed = json!({
                "id": payload["user_id"],
                "name": payload["username"],
                "created_at": chrono::Utc::now().to_rfc3339(),
                "status": "active"
            });
            Ok(transformed)
        },
    );

    let input = MessageBuilder::new()
        .with_topic("user.created")
        .with_payload(json!({
            "user_id": 123,
            "username": "alice",
            "email": "alice@example.com"
        }))
        .build();

    let result = processor.process(input).await.unwrap();
    assert_eq!(result["id"], json!(123));
    assert_eq!(result["name"], json!("alice"));
    assert_eq!(result["status"], json!("active"));
    assert!(result["created_at"].is_string());
}

#[tokio::test]
async fn test_processor_chain() {
    // 创建处理器链：过滤 -> 映射 -> 路由
    let filter = Box::new(MapProcessor::new("filter_to_map", |msg| {
        if msg.payload()["active"].as_bool().unwrap_or(false) {
            msg
        } else {
            // 创建一个标记为跳过的消息
            let mut new_msg = msg;
            let mut payload = new_msg.payload().clone();
            payload["_skip"] = json!(true);
            new_msg.set_payload(payload);
            new_msg
        }
    }));

    let mapper = Box::new(MapProcessor::new("enricher", |mut msg: Message| {
        if !msg.payload()["_skip"].as_bool().unwrap_or(false) {
            let mut payload = msg.payload().clone();
            payload["processed_at"] = json!(chrono::Utc::now().timestamp());
            payload["enriched"] = json!(true);
            msg.set_payload(payload);
        }
        msg
    }));

    let chain = ProcessorChain::new("test_chain")
        .add_processor(filter)
        .add_processor(mapper);

    // 测试活跃消息
    let active_msg = MessageBuilder::new()
        .with_topic("test")
        .with_payload(json!({"active": true, "id": 1}))
        .build();

    let result = chain.process(active_msg).await.unwrap();
    assert_eq!(result.payload()["id"], json!(1));
    assert_eq!(result.payload()["enriched"], json!(true));
    assert!(result.payload()["processed_at"].is_number());
    assert!(!result.payload()["_skip"].as_bool().unwrap_or(false));

    // 测试非活跃消息
    let inactive_msg = MessageBuilder::new()
        .with_topic("test")
        .with_payload(json!({"active": false, "id": 2}))
        .build();

    let result = chain.process(inactive_msg).await.unwrap();
    assert_eq!(result.payload()["id"], json!(2));
    assert_eq!(result.payload()["_skip"], json!(true));
    assert!(result.payload()["enriched"].is_null());
}

#[tokio::test]
async fn test_stateful_processor() {
    let map_processor = MapProcessor::new("test_processor", |msg| msg);
    let stateful_processor = StatefulProcessor::new(map_processor);

    let msg = MessageBuilder::new()
        .with_topic("test")
        .with_payload(json!({}))
        .build();

    // 处理几个消息
    for i in 0..5 {
        let test_msg = MessageBuilder::new()
            .with_topic("test")
            .with_payload(json!({"id": i}))
            .build();
        
        let _ = stateful_processor.process(test_msg).await.unwrap();
    }

    let stats = stateful_processor.get_stats().await;
    assert_eq!(stats.total_processed, 5);
    assert_eq!(stats.successful, 5);
    assert_eq!(stats.failed, 0);
    assert!(stats.avg_processing_time_ms >= 0.0);
    assert!(stats.last_processed.is_some());
}

#[tokio::test]
async fn test_parallel_processor() {
    let map_processor = Box::new(MapProcessor::new("slow_processor", |mut msg: Message| {
        // 模拟慢处理
        std::thread::sleep(Duration::from_millis(10));
        let mut payload = msg.payload().clone();
        payload["processed"] = json!(true);
        msg.set_payload(payload);
        msg
    }));

    let parallel_processor = ParallelProcessor::new("parallel_test", map_processor, 4);

    let messages: Vec<Message> = (0..10)
        .map(|i| {
            MessageBuilder::new()
                .with_topic("test")
                .with_payload(json!({"id": i}))
                .build()
        })
        .collect();

    let start = std::time::Instant::now();
    let results = parallel_processor.process(messages).await.unwrap();
    let duration = start.elapsed();

    assert_eq!(results.len(), 10);
    for (i, result) in results.iter().enumerate() {
        assert_eq!(result.payload()["id"], json!(i));
        assert_eq!(result.payload()["processed"], json!(true));
    }

    // 并行处理应该比顺序处理快
    assert!(duration < Duration::from_millis(100)); // 如果是顺序处理需要100ms+
}

#[tokio::test]
async fn test_processor_error_handling() {
    let error_processor = MapProcessor::new("error_processor", |_msg: Message| {
        panic!("Processing error");
    });

    let msg = MessageBuilder::new()
        .with_topic("test")
        .with_payload(json!({}))
        .build();

    // 这个测试预期会panic，在实际使用中应该用try-catch包装
    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        tokio::runtime::Runtime::new().unwrap().block_on(async {
            error_processor.process(msg).await
        })
    }));

    assert!(result.is_err());
}